package com.navinfo.platform.test;

import com.navinfo.opentsp.platform.location.protocol.common.LCFaultUpload;
import com.navinfo.opentsp.platform.location.protocol.common.LCIncidentAlarmData;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class testKafkaClient_a203 {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "10.30.50.121:9092");//you should write specific ip address rather than localhost
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//StringSerializer/IntegerSerializer/or other self-defined Serializer.
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
//        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("group.id", "MyGroup");
        properties.put("auto.offset.reset","earliest");

        Consumer<String,byte[]> consumer  = new KafkaConsumer<String,byte[]>(properties);
        consumer.subscribe(Arrays.asList("faultUpload"));

        while(true){
            ConsumerRecords<String,byte[]> records = consumer.poll(100);
            for(ConsumerRecord re:records){
                try {

                    LCFaultUpload.FaultUpload faultUpload = LCFaultUpload.FaultUpload.parseFrom((byte[]) (re.value()));
                    System.out.println(faultUpload.getTerminalId());

                }catch (Exception e){
                   e.printStackTrace();
                }
            }
        }
    }
}